package com.rxjava;

import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.subjects.AsyncSubject;

//关于Subject，官方文档的解释是这样的：Subject可以看成是一个桥梁或者代理，在某些ReactiveX实现中（如RxJava），它同时充当了Observer和Observable的角色。因为它是一个Observer，它可以订阅一个或多个Observable；又因为它是一个Observable，它可以转发它收到(Observe)的数据，也可以发射新的数据。
public class AsyncSubjectDemo {

	public static void main(String[] args) {
		//Observer会接收AsyncSubject的```onComplete()``之前的最后一个数据，如果因异常而终止，AsyncSubject将不会释放任何数据，但是会向Observer传递一个异常通知。
		//下例中，Observer只会接收asyncSubject的onCompleted()被调用前的最后一个数据，即“asyncSubject3”，如果不调用onCompleted()，Subscriber将不接收任何数据。
		AsyncSubject<String> asyncSubject = AsyncSubject.create();
        asyncSubject.onNext("asyncSubject1");
        asyncSubject.onNext("asyncSubject2");
        asyncSubject.onNext("asyncSubject3");
        asyncSubject.onComplete();
        asyncSubject.subscribe(new Observer<String>() {
            @Override
            public void onComplete() {
                System.out.println("asyncSubject onCompleted");  //输出 asyncSubject onCompleted
            }

            @Override
            public void onError(Throwable e) {
            	System.out.println("asyncSubject onError");  //不输出（异常才会输出）
            }

            @Override
            public void onNext(String s) {
            	System.out.println("asyncSubject:"+s);  //输出asyncSubject:asyncSubject3
            }

			@Override
			public void onSubscribe(Disposable disposable) {

			}
        });

	}

}
